package com.fwj.storm.sentence;

import org.apache.storm.spout.SpoutOutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichSpout;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Values;
import org.apache.storm.utils.Utils;

import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author 傅文江
 * <p>
 * 时间： 2017-11-07 14:18
 * <p>
 * 描述：制造单词数据源
 * <p>
 * 版权： Copyright © 2017 新道科技股份有限公司
 **/
public class SentenceSpout extends BaseRichSpout {
    private SpoutOutputCollector collector;
    private ConcurrentHashMap<UUID,Values> pending=new ConcurrentHashMap<>();
    private String[] sentences={
            "my dog has fleas",
            "I like cold beverages","the dog ate my homework",
            "donnot have a cow man",
            "I dont think I like fleas" };
    private int index=0;
    @Override
    public void open(Map conf, TopologyContext context, SpoutOutputCollector collector) {
        this.collector=collector;
    }

    @Override
    public void nextTuple() {
        Values tuple = new Values(sentences[index]);
        UUID msgId=UUID.randomUUID();
        this.pending.put(msgId,tuple);
        this.collector.emit(tuple,msgId);
        index++;
        if(index>=sentences.length){
            index=0;
        }
        Utils.sleep(1000);
    }

    public SpoutOutputCollector getCollector() {
        return this.collector;
    }

    @Override
    public void ack(Object msgId) {
        //执行成功移除元数据
        UUID uuid=(UUID)msgId;
        this.pending.remove(uuid);
    }

    @Override
    public void fail(Object msgId) {
        //执行失败重新执行
        UUID uuid=(UUID)msgId;
        this.collector.emit(this.pending.get(uuid),msgId);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields("sentence"));
    }
}
